package com.yanggu.flink.datastream_api.source

import com.yanggu.flink.datastream_api.pojo.Event
import org.apache.flink.streaming.api.functions.source.SourceFunction

import java.util.concurrent.TimeUnit
import scala.util.Random

class ClickSource extends SourceFunction[Event] {

  private val users = Array("Mary", "Alice", "Bob", "Cary")

  private val urls = Array("./home", "./cart", "./fav", "./prod?id=1", "./prod?id=2", "ssdsdd", "fuyvbjfj", "vjvjjkv", "jfjfjjf"
    , "fdjfjfjkfk", "fififififif", "183289489498")

  private var runFlag = true

  override def run(ctx: SourceFunction.SourceContext[Event]): Unit = {
    val usersLength = users.length
    val urlLength = urls.length
    val random = new Random()
    while (runFlag) {
      ctx.collect(Event(users(random.nextInt(usersLength)), urls(random.nextInt(urlLength)), System.currentTimeMillis()))
      TimeUnit.MILLISECONDS.sleep(100L)
    }
  }

  override def cancel(): Unit = {
    runFlag = false
  }

}
